/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.tabletserver;

import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

import com.nhncorp.neptune.common.conf.NConfiguration;

/**
 * 정해진 시간 단위로 Tx수와 평균 수행시간을 기록한다. 
 * @author nhn
 *
 */
public class TabletServerTxStatistics {
  static final int TYPE_PUT = 1;
  static final int TYPE_PUT_FAIL = 2;
  static final int TYPE_PUT_SKIP = 3;
  static final int TYPE_GET = 4;
  
  DecimalFormat format = new DecimalFormat("###,###.##");
  List<TimeStatistics> timeStatistics = new ArrayList<TimeStatistics>();  

  AtomicLong sumPutElapsedTime = new AtomicLong(0);
  AtomicLong sumPut = new AtomicLong(0);

  AtomicLong sumPutFailElapsedTime = new AtomicLong(0);
  AtomicLong sumPutFail = new AtomicLong(0);
  
  AtomicLong sumPutSkip = new AtomicLong(0);

  AtomicLong sumGetElapsedTime = new AtomicLong(0);
  AtomicLong sumGet = new AtomicLong(0);
  
  
  public TabletServerTxStatistics(NConfiguration conf, ThreadGroup threadGroup) {
    final int interval = conf.getInt("tabletServer.tx.statistics.interval", 10) * 60 * 1000;
    Thread collectThread = new Thread(threadGroup, new Runnable() {
      public void run() {
        while(true) {
          try {
            Thread.sleep(interval);
          } catch (InterruptedException e) {
            return;
          }
          
          long putCount = sumPut.get();
          long putElapsedTime = sumPutElapsedTime.get();

          long putFailCount = sumPutFail.get();
          long putFailElapsedTime = sumPutFailElapsedTime.get();

          long putSkipCount = sumPutSkip.get();

          long getCount = sumGet.get();
          long getElapsedTime = sumGetElapsedTime.get();
          
          synchronized(timeStatistics) {
            if(timeStatistics.size() > 5000) {
              int removeNum = timeStatistics.size() - 5000;
              for(int i = 0; i < removeNum; i++) {
                timeStatistics.remove(0);
              }
            }
            timeStatistics.add(new TimeStatistics(System.currentTimeMillis(), getCount, getElapsedTime,
                putCount, putElapsedTime, putFailCount, putFailElapsedTime, putSkipCount));
          }
          sumPut.set(0);
          sumPutElapsedTime.set(0);

          sumGet.set(0);
          sumGetElapsedTime.set(0);
          
          sumPutFail.set(0);
          sumPutFailElapsedTime.set(0);
          
          sumPutSkip.set(0);
        }
      }
    });
    collectThread.start();
  }
  
  public String[] getTxStatistics() {
    synchronized(timeStatistics) {
      if(timeStatistics.size() == 0) {
        return new String[]{"No TxStatistics"};
      } else {
        List<String> result = new ArrayList<String>();
        for(TimeStatistics eachItem: timeStatistics) {
          result.add(eachItem.toString());
        }
        
        return result.toArray(new String[]{});
      }
    }
  }
  
  /**
   * Tx 내역을 추가한다.
   * @param isGet true:get, fasle: put
   * @param elapsedTime
   */
  public void addTx(int txType, long elapsedTime) {
    if(txType == TYPE_GET) {
      sumGet.incrementAndGet();
      sumGetElapsedTime.getAndAdd(elapsedTime / 1000000);
    } else if(txType == TYPE_PUT) {
      sumPut.incrementAndGet();
      sumPutElapsedTime.getAndAdd(elapsedTime / 1000000);
    } else if(txType == TYPE_PUT_FAIL) {
      sumPutFail.incrementAndGet();
      sumPutFailElapsedTime.getAndAdd(elapsedTime / 1000000);
    } else if(txType == TYPE_PUT_SKIP) {
      sumPutSkip.incrementAndGet();
    }
  }
  
  class TimeStatistics {
    boolean type;     //true:get, false: put
    long dateTime;
    long numGet;
    long totalGetElapsedTime;

    long numPut;
    long totalPutElapsedTime;

    long numPutFail;
    long totalPutFailElapsedTime;
    
    long numPutSkip;
    
    public TimeStatistics(long dateTime, 
        long numGet, long totalGetElapsedTime, 
        long numPut, long totalPutElapsedTime, 
        long numPutFail, long totalPutFailElapsedTime,
        long numPutSkip) {
      this.dateTime = dateTime;
      this.numGet = numGet;
      this.totalGetElapsedTime = totalGetElapsedTime;

      this.numPut = numPut;
      this.totalPutElapsedTime = totalPutElapsedTime;
      
      this.numPutFail = numPutFail;
      this.totalPutFailElapsedTime = totalPutFailElapsedTime;
      
      this.numPutSkip = numPutSkip;
    }
    
    public String toString() {
      String result = (new Date(dateTime)).toString();
      
      if(numPut != 0) {
        double avgPutFail = 0;
        if(numPutFail != 0) {
          avgPutFail = (double)totalPutFailElapsedTime / (double)numPutFail;
        }
        result +=( " [put: " +  totalPutElapsedTime + "/" + 
                numPut + " = " + format.format((double)totalPutElapsedTime / (double)numPut) + " ms], " +
                	  " [put fail: " + totalPutFailElapsedTime + "/" + 
                	  numPutFail + " = " + format.format(avgPutFail) + " ms], " +
                		" [put skip:" + numPutSkip + "]");
      } 
      if(numGet != 0) {
        result +=( "[get: " +  totalGetElapsedTime + "/" + 
            numGet + " = " + format.format((double)totalGetElapsedTime / (double)numGet) + " ms] ");
      }
      
      return result;
    }
  }
}